package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.app.BaseTenvApp;
import com.atguigu.realtime.app.function.KeyWordProductUDTF;
import com.atguigu.realtime.app.function.KeyWordUDTF;
import com.atguigu.realtime.common.ConstantTopic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.SQLException;
import java.util.Map;

/**
 * @ClassName: DWSKeyWordProductStatsApp
 * @Description:
 * @Author: kele
 * @Date: 2021/4/28 9:51
 *
 * 商品关键词与点击次数、订单次数、添加购物次数的统计表。
 *
 * 1、从kafka消费product_stats中的数据
 * 2、按照spu字段进行炸裂
 * 3、将click，cart，order字段炸裂
 *
 **/
public class DWSKeyWordProductStatsApp extends BaseTenvApp {

    public static void main(String[] args) {

        new DWSKeyWordProductStatsApp().init(
                40005,
                2,
                "DWSKeyWordProductStatsApp"
        );
    }

    @Override
    protected void run(StreamTableEnvironment tenv) {

        //1.1、读取product_stats中的数据
        tenv.executeSql("create table KeyWordProduct(\n" +
                " spu_name STRING,\n" +
                " click_ct BIGINT,\n" +
                " cart_ct BIGINT,\n" +
                " order_ct BIGINT,\n" +
                " stt STRING,\n" +
                " edt STRING\n" +
                ")with(\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic'='"+ConstantTopic.DWS_PRODUCT_STATS+"',\n" +
                " 'properties.bootstrap.servers'='hadoop162:9092,hadoop163:9092,hadoop164:9092',\n" +
                " 'properties.group.id'='DWSProvinceStatsSqlApp',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n" +
                ")");



        //1.2、建表，与clickhouse关联
        tenv.executeSql("create table keyword_stats_2021(" +
                "   stt string," +
                "   edt string," +
                "   keyword string," +
                "   source string," +
                "   ct bigint," +
                "   ts bigint," +
                "   PRIMARY KEY (stt, edt, keyword, source) NOT ENFORCED" +    //主键字段，和clickhouse中的orderby字段相同（clickhouse中按照orderby的字段进行聚合
                ")with(" +
                "   'connector' = 'clickhouse', " +
                "   'url' = 'clickhouse://hadoop162:8123', " +
                "   'database-name' = 'gmall2021', " +
                "   'table-name' = 'keyword_stats_2021'," +
                "   'sink.batch-size' = '1', " +
                "   'sink.flush-interval' = '1000', " +
                "   'sink.max-retries' = '3' " +
                ")"
        );

        //2.1  注册炸裂函数，炸裂spu_name字段
        tenv.createTemporaryFunction("ik_analyze", KeyWordUDTF.class);

        Table t1 = tenv.sqlQuery("select \n" +
                " keyword,\n" +
                " click_ct,\n" +
                " cart_ct,\n" +
                " order_ct,\n" +
                " stt,\n" +
                " edt\n" +
                "from KeyWordProduct,\n" +
                "lateral table (ik_analyze(spu_name)) as T(keyword)");

        tenv.createTemporaryView("t1",t1);

        //2.2  需要先按照关键字进行聚合，否则在click_house中会去重
        Table t2 = tenv.sqlQuery("select \n" +
                " keyword,\n" +
                " sum(click_ct) click_ct,\n" +
                " sum(cart_ct) cart_ct,\n" +
                " sum(order_ct) order_ct,\n" +
                " stt,\n" +
                " edt \n" +
                "from t1\n" +
                "group by stt,edt,keyword");

        tenv.createTemporaryView("t2",t2);
        tenv.createTemporaryFunction("kwp", KeyWordProductUDTF.class);

        Table t3 = tenv.sqlQuery("select \n" +
                " stt,\n" +
                " edt,\n" +
                " keyword,\n" +
                " source,\n" +
                " ct,\n" +
                " unix_timestamp()*1000 ts\n" +
                "from t2,\n" +
                "lateral table (kwp(click_ct,cart_ct,order_ct))");

        //tenv.sqlQuery("select * from " + t3).execute().print();

        t3.executeInsert("keyword_stats_2021");

    }
}





